package cn.com.tarotframework.server.workflow.service;

import cn.com.tarotframework.common.Comments;
import cn.com.tarotframework.config.ThreadPoolTaskExecuteConfig;
import cn.com.tarotframework.server.workflow.bo.ClusterRateFlow;
import cn.com.tarotframework.server.workflow.bo.FlowNodeAssemble;
import cn.com.tarotframework.server.workflow.bo.FlowNodeTask;
import cn.com.tarotframework.server.workflow.eu.FlowStatus;
import cn.com.tarotframework.server.workflow.po.FlowLine;
import cn.com.tarotframework.server.workflow.po.FlowNode;
import cn.com.tarotframework.utils.FlowNodeUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Slf4j
@Service
public class BusinessService {

    private List<FlowNode> flowNodes;
    private List<FlowLine> flowLines;
    private List<ClusterRateFlow> clusterRateFlows;
    private final IFlowNodeService flowNodeService;

    public static ExecutorService service = Executors.newFixedThreadPool(10);


    private final ThreadPoolTaskExecuteConfig executeConfig;

    public BusinessService(IFlowNodeService flowNodeService, ThreadPoolTaskExecuteConfig executeConfig) {
        this.flowNodeService = flowNodeService;
        this.executeConfig = executeConfig;
    }


    /**
     * 执行节点流程
     * @return List<FlowNode>
     */
    public List<FlowNode> execute( List<FlowNode> flowNodeList, List<FlowLine> flowLineList, List<ClusterRateFlow> clusterRateFlowList) throws ExecutionException, InterruptedException {
        this.flowNodes = flowNodeList;
        this.flowLines = flowLineList;
        this.clusterRateFlows = clusterRateFlowList;
        //检测执行节点的流量是否超过集群设定的流量，如果超过了，将超过的节点信息弹回客户端，由客户端调整集群流量或节点流量
        List<FlowNode> rates = exceedQuotaRateFlow(this.flowNodes);
        if(!CollectionUtils.isEmpty(rates)) {
            rates.forEach(System.out::println);
            return flowNodes;
        }
        // 开始执行：判断是否之前赞停过,为空则从头开始执行，否则获取成功的节点往下执行
        List<FlowNode> success = this.flowNodes.stream().filter( node -> node.getStatus() == FlowStatus.EXECUTE_SUCCESS).collect(Collectors.toList());
        if(CollectionUtils.isEmpty(success)) {
            recurrenceExecuteFlowNode(this.flowNodes.get(0));
        }else {
            success.forEach(this::recurrenceExecuteFlowNode);
        }
        System.out.println("#################################### 最终结果 ####################################");
        flowNodes.forEach(System.out::println);
        return this.flowNodes;
    }


    /**
     * 检测集合节点是否在当前集群中超标
     * @param flowNodeList 节点集合
     * @return List<FlowNode>
     */
    private List<FlowNode> exceedQuotaRateFlow(List<FlowNode> flowNodeList) {
        return this.clusterRateFlows.stream().filter( rate -> rate.getRateFlow() > 0).flatMap(
            cluster -> flowNodeList.stream().filter( node -> node.getClusterId().equals(cluster.getClusterId()) && node.getParallel() > cluster.getRateFlow())
        ).collect(Collectors.toList());
    }

    /**
     * 开始执行
     * @param flowNode 流程业务对象
     */
    private void recurrenceExecuteFlowNode(FlowNode flowNode) {
        // 判断是否为根节点，如果是，执行
        if(flowNode.getId().equals("node-start-0")) {
            this.flowNodeService.executeComplete(flowNode).join();
        }
        // 获取当前节点的所有子节点
        List<FlowNode> childFlowNodes = FlowNodeUtil.adjacentFlowNode(flowNodes, flowLines, flowNode, Comments.EXECUTE_AFTER_NODE)
                .stream().filter( node -> node.getStatus() == FlowStatus.EXECUTE_WAIT).collect(Collectors.toList());
        if(!CollectionUtils.isEmpty(childFlowNodes)) {
            // 通过算法(集群流量，节点流量)，过滤、查找、分解待执行节点集合，提取真正将要执行的集合，和待执行集合
            FlowNodeAssemble<FlowNode> flowNodeAssemble = this.flowNodeAssemble(childFlowNodes);
            // 执行节点集合
            this.flowNodeService.executeComplete(
                flowNodeAssemble.getExecuteFlowNodes().stream().filter(node -> node.getStatus() == FlowStatus.EXECUTE_WAIT).collect(Collectors.toList())
            ).join();
            // 执行等待节点集合
            recurrenceWaitFlowNodes(flowNodeAssemble);
            // 合并集合，提取所有子节点,将这些节点进行当前方法递归处理
            List<FlowNode> flowNodes = FlowNodeUtil.mergeFlowNodes(this.flowNodes, this.flowLines, flowNodeAssemble.getExecuteFlowNodes(),flowNodeAssemble.getWaitFlowNodes());
            // 递归执行
            flowNodes.forEach(this::recurrenceExecuteFlowNode);
            System.out.println("-------------------------子节点递归数据 start ------------------------------");
            flowNodes.forEach(System.out::println);
            System.out.println("-------------------------子节点递归数据 end --------------------------------");
        }
    }

    /**
     * 递归执行待执行节点集合，直至该集合为空，结束递归
     * @param flowNodeAssemble 流程业务对象
     */
    private void recurrenceWaitFlowNodes(FlowNodeAssemble<FlowNode> flowNodeAssemble) {
        List<FlowNode> waitFlowNodes = flowNodeAssemble.getWaitFlowNodes();
        if(CollectionUtils.isEmpty(waitFlowNodes)) {
            System.out.println("################## 子节点集合递归，集合为空，结束内层递归调用 ##################");
            return;
        }
        // 流量与节点计算（区分执行节点集合，待执行节点集合）
        flowNodeAssemble = this.executeFlowNodes(waitFlowNodes);
        System.out.println("---------------------外层，要执行的节点集合 --------------------------------");
        flowNodeAssemble.getExecuteFlowNodes().forEach(System.out::println);
        System.out.println("--------------------------------------------------------------------------");
        System.out.println("---------------------外层，待执行的节点集合 --------------------------------");
        flowNodeAssemble.getWaitFlowNodes().forEach(System.out::println);
        System.out.println("--------------------------------------------------------------------------");
        // 执行节点开始执行
        FlowNodeAssemble<FlowNode> finalFlowNodeAssemble = flowNodeAssemble;
        this.flowNodeService.executeComplete(
            finalFlowNodeAssemble.getExecuteFlowNodes().stream().filter(node -> node.getStatus() == FlowStatus.EXECUTE_WAIT).collect(Collectors.toList())
        ).join();
        recurrenceWaitFlowNodes(flowNodeAssemble);
    }

    /**
     * 通过算法(集群流量，节点流量)，过滤、查找、分解待执行节点集合，提取真正将要执行的集合，和待执行集合
     * @param currentFlowNodes 当前节点下子节点集合
     * @return FlowNodeAssemble<FlowNode>
     */
    private FlowNodeAssemble<FlowNode> flowNodeAssemble(List<FlowNode> currentFlowNodes) {
        System.out.println("------------------1：当前节点，将要执行的所有子节点 --------------------------");
        currentFlowNodes.forEach(System.out::println);
        System.out.println("-------------------------------------------------------------------------");
        // 1：在当前集合中，根据当前集合获取父节点集合 将当前结合与父节点集合相比，判断父节点为：EXECUTE_WAIT的对应的 子节点集合的 节点集合
        List<FlowNode> childFlowNodes = FlowNodeUtil.searchCurrentListChildFlowNodes(this.flowNodes, this.flowLines, currentFlowNodes);
        // 3：从当前将要执行节点集合中，删除从中提取的子节点
        List<FlowNode> executeNodes = FlowNodeUtil.cleanExecuteNode(currentFlowNodes, childFlowNodes);
        // 4: 根据集群流量，计算出当前将要执行节点集合中，可以真正进行执行的节点集合
        FlowNodeAssemble<FlowNode> assembles = this.executeFlowNodes(executeNodes);
        // 5: 将之前筛选出来的子节点集合，合并到待执行节点集合中
        assembles.setWaitFlowNodes(FlowNodeUtil.mergeFlowNodes(assembles.getWaitFlowNodes(), childFlowNodes));
        System.out.println("------------------------内层：要执行的节点集合-----------------------------");
        assembles.getExecuteFlowNodes().forEach(System.out::println);
        System.out.println("------------------------------------------------------------------------");
        System.out.println("------------------------内层：待执行的节点集合-----------------------------");
        assembles.getWaitFlowNodes().forEach(System.out::println);
        System.out.println("------------------------------------------------------------------------");
        return assembles;
    }


    /**
     * 区分真正将要执行的节点，和待执行节点
     * @param currentFlowNodes 当前执行节点结合
//     * @param parallel 集群总流量
     * @return  FlowNodeAssemble<FlowNode>
     */
    private FlowNodeAssemble<FlowNode> executeFlowNodes(List<FlowNode> currentFlowNodes) {
        FlowNodeAssemble<FlowNode> flowNodeAssemble = new FlowNodeAssemble<>(new ArrayList<>(), new ArrayList<>());
        // 提取当前节点集合中，节点对应的所有父级节点，判断状态是否为 EXECUTE_WAIT，如果是，先删除，目的优先级放后排
        List<FlowNode> temporaryTroopNode = FlowNodeUtil.searchCurrentListChildFlowNodes(this.flowNodes, this.flowLines,currentFlowNodes);
        // 暂时清除该节点
        FlowNodeUtil.cleanExecuteNode(currentFlowNodes, temporaryTroopNode);
        System.out.println("-----------------以下节点还有父节点没有执行完成, 先等待-------------------------");
        temporaryTroopNode.forEach(System.out::println);
        System.out.println("--------------------------------------------------------------------------");
        if(currentFlowNodes.size() > 0) {
            // 然后根据集群ID进行分组. 遍历分组集合，计算每个集群对应节点，是否可以参与执行
            currentFlowNodes.stream().collect(Collectors.groupingBy(FlowNode::getClusterId)).forEach((clusterId, flowNodes) -> {
                AtomicInteger count = new AtomicInteger(), index = new AtomicInteger();
                int clusterRateFlow = this.clusterRateFlows.stream().collect(Collectors.groupingBy(ClusterRateFlow::getClusterId)).get(clusterId).get(0).getRateFlow();
                // 当前集合提取，与之相对应集群的节点
                List<FlowNode> executeFlowNodes = flowNodes.stream().filter(node -> clusterRateFlow - count.get() >= node.getParallel()).peek(node -> {
                    count.set(node.getParallel() + count.get());
                    index.getAndIncrement();
                }).collect(Collectors.toList());
                // 提取待执行节点
                List<FlowNode> waitFlowNodes = flowNodes.stream().skip(index.get()).filter(node -> node.getStatus() == FlowStatus.EXECUTE_WAIT).collect(Collectors.toList());
                flowNodeAssemble.getExecuteFlowNodes().addAll(clusterRateFlow <= 0 ? flowNodes : executeFlowNodes);
                flowNodeAssemble.getWaitFlowNodes().addAll(clusterRateFlow <= 0 ? new ArrayList<>() : waitFlowNodes);
            });
        }
        return flowNodeAssemble;
    }
}
